package com.java.concurrent;

/**
 * @program: java_basic_knowledge
 * @description:
 * @author: CaoYong
 * @create: 2021-04-23 10:08
 **/

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.Callable;

/**
 * 线程协作场景：
 * 1. 生产者/消费者： 生产者，消费者线程
 * 2. 同时开始： 多个线程同时开始
 * 3. 等待结束： 主线程将任务拆分为多个子任务并为每个任务创建一个线程，主线程在执行其他任务前需要都等待每个子进程执行结束
 * 4. 异步结果: 将子线程封装为异步调用，异步调用马上返回但返回的不是最终结果，而是一个Future对象，通过它可以在之后获得最终的结果
 * 5. 集合点：每个线程负责一部分计算，然后在集合点等待其他线程完成，等所有线程到齐后，交换数据和执行结果，再往下执行
 */
public class ThreadCollaborationKnowledge {

    public class WhatIsWaitAndNotify {

        // java在Object类而不是Thread类定义了线程协作的基本方法及wait()和notify()，因此每个对象都可以调用这些方法
        // 每个Object对象除了用于锁的等待队列还有条件队列，用于线程间协作
        // 调用wait()方法及将当前线程放到条件队列上并阻塞，表示当前线程无法执行需要等待一个条件，而这个条件需要其他线程提供
        // 其他线程改变了条件后通过调用notify()方法通知
        // notify()方法即从条件队列中选择一个线程将其从队列中移除并唤醒
        // wait和notify被不同的线程调用，但共享相同的锁和条件等待队列，围绕着一个共享的条件变量进行协作
        // 每个对象都有一把锁和等待队列，未得到锁的线程会进入等待队列
        // 除此之外每个对象还有另一个等待队列表示条件队列，用于线程间的协作
        // wait()会将当前线程放到条件队列上并阻塞，表示线程无法进行下去必须要等待一个条件，而这个条件它自己无法改变
        // 其他线程在改变完条件后通过调用notify()方法通知，及从条件队列中选择一个线程将其从队列中移除并唤醒
        public class Object {
//            public final native void notify();
//            public final native void notifyAll();
//            public final native void wait(long timeout) throws InterruptedException;
//            public final void wait() throws InterruptedException;
        }
    }

    public static class HowToUseWaitAndNotify {

        public static class WaitThread extends Thread {
            private volatile boolean fire = false;

            @Override
            public void run() {
                try {
                    synchronized (this) {
                        while (! fire) {
                            wait();
                        }
                    }
                    System.out.println("fired");
                } catch (Exception e) {

                }
            }

            public synchronized void fire() {
                this.fire = true;
                notify();
            }
        }

        /**
         * 两个线程都要访问协作变量fire容易产生竞态条件，所以相关代码都需要被synchronized保护
         * 实际上wait/notify方法只能在synchronized代码块内被调用，否则会抛出illegalMonitor-StateException
         * 虽然在synchronized方法内，但是调用wait()方法时线程会释放对象锁
         * 1.把当前线程放入条件等待队列，释放对象锁，阻塞等待，线程状态变为WAITING或TIMED_WAITING
         * 2.当时间到或者被其他线程调用notify()方法，从条件队列中移除，此时需要重新竞争锁对象
         * 3.如果能竞争获得锁，线程状态变为RUNNABLE并从wait调用中返回，否则加入锁的等待队列，线程状态变为BLOCKED, 获得锁后才会从wait中返回
         *
         */
        public static void main(String[] args) throws InterruptedException {
            WaitThread waitThread = new WaitThread();
            waitThread.start(); //waitThread线程wait()
            Thread.sleep(1000);
            System.out.println("fire");
            waitThread.fire(); //main线程调用fire()方法进行notify()
        }

    }


    // 生产者消费者模式
    public static class WhatIsProducerAndConsumerPattern {

        static class BlockingQueue<E> {
            private Queue<E> queue = null;
            private int limit;

            public BlockingQueue(int limit) {
                this.limit = limit;
                queue = new ArrayDeque<>(limit);
            }

            // put()方法给生产者使用，往队列上放数据满了就wait，放完之后调用notifyAll通知消费者
            public synchronized void put(E e) throws InterruptedException {
                while (queue.size() == limit) {
                    wait();
                }

                queue.add(e);
                notifyAll();
            }

            // take()给消费者使用，从队列中拉取数据，如果队列为空就wait, 取完数据后调用notifyAll通知生产者
            // put()和take()都调用了wait()但是其目的并不同, 但是他们都会加入相同的条件等待队列，而notify只能唤醒一个线程，所以这里要用notifyAll
            // 只能有一个条件等待队列时java wait/notify的局限，显示锁和条件可以解决该问题
            public synchronized E take() throws InterruptedException {
                while (queue.isEmpty()) {
                    wait();
                }
                E e = queue.poll();
                notifyAll();
                return e;
            }
        }

        static class Producer extends Thread {
            BlockingQueue<String> queue;

            public Producer(BlockingQueue<String> queue) {
                this.queue = queue;
            }

            @Override
            public void run() {
                int num = 0;
                try {
                    while (true) {
                        String task = String.valueOf(num);
                        queue.put(task);
                        System.out.println("produce task:" + task);
                        num ++;
                        Thread.sleep((int)(Math.random() * 100));
                    }
                } catch (Exception e) {

                }
            }
        }

        static class Consumer extends Thread {
            BlockingQueue<String> queue;

            public Consumer(BlockingQueue<String> queue) {
                this.queue = queue;
            }

            @Override
            public void run() {
                try {
                    while (true) {
                        String task = queue.take();
                        System.out.println("handle task:" + task);
                        Thread.sleep((int) (Math.random() *100));
                    }
                } catch (Exception e) {

                }
            }
        }

        public static void main(String[] args) {
            BlockingQueue<String> queue = new BlockingQueue<>(10);
            new Producer(queue).start();
            new Consumer(queue).start();
        }
    }


    // 同时开始
    public static class WhatIsConcurrentlyStartPattern {
        static class FireFlag {
            private volatile boolean fired = false;

            public synchronized void waitForFire() throws InterruptedException {
                while (!fired) {
                    wait();
                }
            }

            public synchronized void fire() {
                this.fired = true;
                notifyAll();
                System.out.println(Thread.currentThread().getName() + ":" + "fired");
            }
        }

        static class Racer extends Thread {
            FireFlag fireFlag;

            public Racer(FireFlag fireFlag) {
                this.fireFlag = fireFlag;
            }

            @Override
            public void run() {

                try {
                    this.fireFlag.waitForFire();
                    System.out.println("start run:" + Thread.currentThread().getName() + ":" + System.currentTimeMillis());
                } catch (InterruptedException e) {
                }
            }

            public static void main(String[] args) throws InterruptedException {
                int num = 10;
                FireFlag fireFlag = new FireFlag();
                Thread[] threads = new Thread[num];
                for (int i = 0; i < num; i++) {
                    threads[i] = new Racer(fireFlag);
                    threads[i].start();
                }
                Thread.sleep(1000);
                fireFlag.fire(); //fire()发布信号后，所有wait()的线程同时执行
            }
        }
    }


    // 等待结束
    public static class WhatIsWaitEnd {

        static class MyLatch {
            private static int count;

            public MyLatch(int count) {
                this.count = count;
            }

            public synchronized void await() throws InterruptedException {
                while (count > 0) {
                    wait();
                }
            }

            public synchronized void countDown() {
                count --;
                if (count <= 0) {
                    notifyAll();
                }
            }
        }

        static class Worker extends Thread {
            MyLatch latch;

            public Worker(MyLatch latch) {
                this.latch = latch;
            }

            @Override
            public void run() {
                try {
                    Thread.sleep((int) (Math.random() * 1000));
                    this.latch.countDown();
                    System.out.println(Thread.currentThread().getName() + ": finish");
                } catch (InterruptedException e) {

                }
            }
        }

        public static void main(String[] args) throws InterruptedException{
            int workerNum = 100;
            MyLatch latch = new MyLatch(workerNum);
            Worker[] workers = new Worker[workerNum];
            for (int i = 0; i < workerNum; i++) {
                workers[i] = new Worker(latch);
                workers[i].start();
            }

            latch.await(); //主线程等待直到所有线程执行结束后再执行
            System.out.println("collect worker results");
        }
    }

    // 异步调用通常返回一个称为Future的对象，通过它可以获得最终的结果
    public static class WhatIsAsynchronousPattern {
        public interface MyFuture<V> {
            //get()方法返回真正的执行结果，如果结果还没有计算完成，get方法会阻塞直达计算完成，如果调用过程中抛出异常则get方法抛出异常
            V get() throws Exception;
        }

        // 定义执行类，执行子任务并返回异步结果
        public static class MyExecutor {
            public <V> MyFuture<V> execute(final Callable<V> task) {
                final Object lock = new Object();
                final ExecuteThread<V> thread = new ExecuteThread<>(task, lock);
                thread.start();
                MyFuture<V> future = new MyFuture<V>() {
                    @Override
                    public V get() throws Exception {
                        synchronized (lock) {
                            while (!thread.isDone()) { // 如果方法没有执行完则一直阻塞，执行完成则返回结果
                                try {
                                    lock.wait();
                                } catch (InterruptedException e) {

                                }
                            }
                        }
                        if (thread.getException() != null) {
                            throw  thread.getException();
                        }
                        return thread.getResult();
                    }
                };
                return future;
            }
        }

        public static class ExecuteThread<V> extends Thread {
            private V result = null;
            private Exception exception = null;
            private boolean done = false;
            private Callable<V> task;
            private Object lock;

            public ExecuteThread(Callable<V> task, Object lock) {
                this.task = task;
                this.lock = lock;
            }

            @Override
            public void run() {
                try {
                    result = task.call();
                } catch (Exception e) {
                    exception = e;
                } finally {
                    synchronized (lock) {
                        done = true;
                        lock.notifyAll(); //任务执行结束进行通知
                    }
                }
            }

            public V getResult() {
                return result;
            }

            public Exception getException() {
                return exception;
            }

            public boolean isDone() {
                return done;
            }
        }

        public static void main(String[] args) {
            MyExecutor executor = new MyExecutor();
            // 定义子任务
            Callable<Integer> subTask = new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    int mills = (int) (Math.random() * 1000);
                    Thread.sleep(mills);
                    return mills;
                }
            };

            // 异步调用子任务，返回一个MyFuture对象
            MyFuture<Integer> future = executor.execute(subTask);

            try {
                Integer result = future.get(); //获取异步调用的结果
                System.out.println(result);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 集合点模式
    public static class WhatIsAssemblePointPattern {

        static class AssemblePoint {
            private int n;

            public AssemblePoint(int n) {
                this.n = n;
            }

            public synchronized void await() throws InterruptedException {
                if (n > 0) {
                    n --;
                    if (n == 0) {
                        notifyAll();
                    } else {
                        while (n != 0) {
                            wait();
                        }
                    }
                }
            }
        }

        public static class Tourist extends Thread{
            AssemblePoint ap;

            public Tourist(AssemblePoint ap) {
                this.ap = ap;
            }

            @Override
            public void run() {
                try {
                    Thread.sleep((int) (Math.random() * 1000));
                    ap.await();
                    System.out.println("arrived");
                } catch (Exception e) {

                }
            }
        }

        public static void main(String[] args) {
            int num = 10;
            Tourist[] tourists = new Tourist[num];
            AssemblePoint ap = new AssemblePoint(num);

            for (int i = 0; i < num; i++) {
                tourists[i] = new Tourist(ap);
                tourists[i].start();
            }
        }
    }





}
